#include "../include/HPLCProxy.h"
#include "../../../include/mosquitto.h"
#include "../../../src/common/include/common.h"
#include "../../logger/include/CLogger.h"

//向HPLC添加设备档案
#define  APPEND_DEVARCHIVES_TOPIC  "pemd/set/request/ccoRouter/acqFiles"
//HPLC添加档案响应
#define  HPLC_APPENDARCHIVES_RSP_TOPIC  "ccoRouter/set/response/pemd/acqFiles"
//清除档案
#define  CLEAR_DEVARCHIVES_TOPIC  "pemd/set/request/ccoRouter/clearAcqFiles"
//清除档案响应
#define  HPLC_CLEARARCHIVES_RSP_TOPIC "ccoRouter/set/response/pemd/clearAcqFiles"
//从节点监控
#define  SALVENODE_CTRL_TOPIC  "pemd/get/request/ccoRouter/monitorNode"
//从节点监控响应
#define  HPLC_SLAVECTRL_RSP_TOPIC "ccoRouter/get/response/pemd/monitorNode"
//并发数据采集
#define  CONCURRENT_ACQDATA_TOPIC  "pemd/get/request/ccoRouter/concurrent"
//采集数据的响应
#define  HPLC_ACQDATA_RSP_TOPIC   "ccoRouter/get/response/pemd/concurrent"
//数据上报
#define  HPLC_REPORT_DATA_TOPIC  "ccoRouter/notify/spont/pemd/reportConcurrent"
//频点设置
#define  HPLC_FREQSET_TOPIC  "pemd/set/request/ccoRouter/hplcFreq"
//频点设置响应
#define  HPLC_FRQSET_RSP_TOPIC "ccoRouter/set/response/pemd/hplcFreq"

HPLCProxy::HPLCProxy(int freqPt) noexcept
{
    m_freqPt = freqPt;
    m_req_id = 0;
    m_is_receiving = false;
    m_rsp_len = 0;
    m_rsp = nullptr;

    init_mqtt();
}


HPLCProxy::~HPLCProxy()
{
}


void HPLCProxy::init_mqtt()
{
    m_mosquitto_manager = std::make_shared<CMosquittoManager>();
    m_mosquitto_manager->init(this, recv_msg_callback);
}

int HPLCProxy::connect(const char *host, int port, const char * username, const char *password, int keepalive /* = 5000 */)
{
    if ((NULL == host) || (1 > strlen(host))){
        return MOSQ_ERR_INVAL;
    }
    int ret = m_mosquitto_manager->connect(host, port, username, password, keepalive);
    if (MOSQ_ERR_SUCCESS == ret){
        m_mosquitto_manager->start();
    }
    else{
        m_mosquitto_manager->stop();
    }
    CLogger::getInstance()->log_i("HPLCProxy[host=%s,port=%d] connect  ret =%d",host,port,ret);
    
    return ret;
}

void HPLCProxy::hplc_freqset()
{
    m_req_id++;
    std::string msg = "";
    std::string topic = HPLC_FREQSET_TOPIC;

    m_hpm.get_hplc_freqset_reqmsg(m_req_id, m_freqPt, msg);
    CLogger::getInstance()->log_i("hplc_freqset: %s", msg.c_str());
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("ccoRouter set freq point[%s] error=%d.", topic.c_str(), ret);
    }
}

//订阅主题
int HPLCProxy::subscribe_topic()
{
    //频点设置响应主题
    std::string topic = HPLC_FRQSET_RSP_TOPIC;
    int mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //添加档案响应主题
    topic = HPLC_APPENDARCHIVES_RSP_TOPIC;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //清除档案响应主题
    topic = HPLC_CLEARARCHIVES_RSP_TOPIC;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //从节点监控响应主题
    topic = HPLC_SLAVECTRL_RSP_TOPIC;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //采集数据控响应主题
    topic = HPLC_ACQDATA_RSP_TOPIC;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    //数据上报响应主题
    topic = HPLC_REPORT_DATA_TOPIC;
    mossub = m_mosquitto_manager->subscribe(topic.c_str());
    if (mossub != MOSQ_ERR_SUCCESS){
        CLogger::getInstance()->LOG_E("subscribe topic[%s] error=%d.", topic.c_str(), mossub);
        return mossub;
    }

    return MOSQ_ERR_SUCCESS;
}

int HPLCProxy::recv_msg_callback(CALLBACK_TYPE cb_type, void *data, int data_len, int result, void *arg)
{
    CLogger::getInstance()->log_d("HPLCProxy callback function(type: %d)", cb_type);
    if ((NULL == data) || (NULL == arg)){
        CLogger::getInstance()->log_e("HPLCProxy callback function data or arg is null");
        return -1;
    }

    HPLCProxy* hpp = (HPLCProxy*)arg;
    if (CALLBACK_TYPE_MESSAGE == cb_type){
        
        if (NULL != data){
            struct mosquitto_message *msg = (struct mosquitto_message *)data;
            std::string  strtopic = msg->topic;
            std::string payload = (char *)msg->payload;
            
            CLogger::getInstance()->log_i("recv_msg_callback recv data:topic=%s\n payload=%s", strtopic.c_str(), payload.c_str());
            if (0 < hpp->parse_rsptopic(strtopic, payload))
            {
                //发送收到数据信号
                hpp->finish_signal();
            }
            
        }
    }
    else if (CALLBACK_TYPE_CONNECTED == cb_type){
        hpp->subscribe_topic();
        //ToDO:hplc空载时，频点设置后添加档案 Failure
        //hpp->hplc_freqset();   
    }

    return 0;
}

int HPLCProxy::parse_rsptopic(std::string topic, std::string payload)
{
    std::string strstatus;
    char recvbuf[512] = { 0 };
    int  recvlen = 0;

    //并发采集请求响应
    if (0 == topic.compare(HPLC_ACQDATA_RSP_TOPIC))
    {
        if (0 < m_hpm.get_hplc_comm_rspdata(payload, strstatus))
        {
            m_rsp_len = strstatus.length();
            memcpy(m_rsp, strstatus.c_str(), m_rsp_len);
        }
        CLogger::getInstance()->log_i("parse_rsptopic  req rspdata: %s", strstatus.c_str());
        memset(m_rsp, 0, sizeof(m_rsp));
        m_rsp_len = 0;
        return 0;   //采集数据为异步，返回0，在外部不结束等待，收到数据上报才结束
    }
    //数据上报
    if (0 == topic.compare(HPLC_REPORT_DATA_TOPIC))
    {        
        std::string  slaveAddr;
        int ret = m_hpm.get_hplc_acqdata_rspmsg(payload, slaveAddr, recvbuf, recvlen);
        if (0 < ret)
        {
            memcpy(m_rsp, recvbuf, recvlen);
            m_rsp_len = recvlen;
            OutputDecodedata(recvbuf, recvlen);
        }
        return 1;  //返回1，在外部结束等待
    }
    //从节点监控
    if (0 == topic.compare(HPLC_SLAVECTRL_RSP_TOPIC))
    {
        if (0 < m_hpm.get_hplc_monitor_rspdata(payload, recvbuf, recvlen,m_rsptoken))
        {
            if (m_rsptoken.compare(m_reqtoken.c_str()) ==0)
            {
                memcpy(m_rsp, recvbuf, recvlen);
                m_rsp_len = recvlen;
                OutputDecodedata(recvbuf, recvlen);
            }
            else{//发送和接收token不一致
                CLogger::getInstance()->LOG_W("parse_rsptopic msg not matching! send token[%s] != recv token[%s].", m_reqtoken.c_str(),m_rsptoken.c_str());
                uint64_t  cur_timestamp = get_timestamp();
                if (cur_timestamp - m_send_timestamp < 40000)  //小于wait_recv_finished中的超时时间
                {
                    return  0;  //返回0，外部不发信号，继续等待
                }
            }
            
        }else{
            m_rsp_len = 0;
        }
        return 1;
    }
    //添加档案、清除档案json格式相同，同用一个解析
    if ((0==topic.compare(HPLC_APPENDARCHIVES_RSP_TOPIC)) ||
        (0 == topic.compare(HPLC_CLEARARCHIVES_RSP_TOPIC)))
    {
        if (0 < m_hpm.get_hplc_comm_rspdata(payload, strstatus))
        {
            m_rsp_len = strstatus.length();
            memcpy(m_rsp, strstatus.c_str(), m_rsp_len);
        }  
        CLogger::getInstance()->log_i("[%s] parse_rsptopic status: %s",topic.c_str(), strstatus.c_str());
        return 1;
    }

    //频点设置返回
    if (0 == topic.compare(HPLC_FRQSET_RSP_TOPIC))
    {
        if (0 < m_hpm.get_hplc_comm_rspdata(payload, strstatus))
        {
            CLogger::getInstance()->log_i("[%s] parse_rsptopic status: %s", topic.c_str(), strstatus.c_str());
        }     
    }

    return 0;
}

void HPLCProxy::finish_signal(){
    m_thread_notify.lock();
    m_is_receiving = false;
    m_thread_notify.signal();
    m_thread_notify.unlock();
}

void HPLCProxy::wait_recv_finished(){
    //最长等待 4S *10
    int ms = 4000;
    int n = 0;
    while (m_is_receiving){
        n++;
        if (n > 10){
            break;
        }
        m_thread_notify.timewait(ms);
    }
    if (n > 10){
        m_rsp_len = 0;
    }
}

int HPLCProxy::slavenode_ctrl(const std::string& slaveAddr, uint8_t *sendBuf, int buf_len, uint8_t* rsp, int& rsplen)
{
    m_thread_notify.lock();
    m_rsp = rsp;
    m_req_id++;
    std::string msg = "";
    // std::string topic = SALVENODE_CTRL_TOPIC;
    std::string topic = CONCURRENT_ACQDATA_TOPIC;

    m_hpm.get_hplc_monitor_reqmsg(m_req_id, slaveAddr, sendBuf, buf_len, msg);

    CLogger::getInstance()->log_i("slavenode_ctrl send: %s", msg.c_str());
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
        rsplen = m_rsp_len;
    }
    else {
        rsplen = 0;
        CLogger::getInstance()->LOG_E("ccoRouter salve node ctrl[%s] error=%d.", topic.c_str(), ret);
    }
    m_thread_notify.unlock();

    return rsplen;
}

int HPLCProxy::add_hplcarchives(std::list<std::string> lstDevAdds, uint8_t* rsp, int& rsplen)
{
    m_thread_notify.lock();

    m_req_id++;
    m_rsp = rsp;
    std::string msg ="";
    std::string topic = APPEND_DEVARCHIVES_TOPIC;

    std::list<ST_DEV_ARCHIVES> lstarchives;
    for (auto devAddr = lstDevAdds.begin(); devAddr != lstDevAdds.end(); devAddr++)
    {
        ST_DEV_ARCHIVES stArchives;
        stArchives.acqAddr = *devAddr;
        stArchives.proType = "02";
        lstarchives.push_back(stArchives);
    }    

    m_hpm.get_hplc_addarchives_reqmsg(m_req_id,lstarchives,msg);
    CLogger::getInstance()->log_i("add_hplcarchives send: %s", msg.c_str());
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
    }
    else {
        rsplen = 0;
        CLogger::getInstance()->LOG_E("ccoRouter add archives[%s] error=%d.", topic.c_str(), ret);
    }
    m_thread_notify.unlock();

    return rsplen;
}

int HPLCProxy::send_msg(uint8_t *req, int req_len, uint8_t *rep, int &rep_len, uint32_t timeout)
{
    m_thread_notify.lock();

    m_req_id++;
    m_rsp = rep;
    std::string msg = "";
    // std::string topic = SALVENODE_CTRL_TOPIC;
    std::string topic = CONCURRENT_ACQDATA_TOPIC;

    m_hpm.get_hplc_monitor_reqmsg(m_req_id, c_slave_addr, req, req_len, msg);
    CLogger::getInstance()->log_i("HPLCProxy,req_id=%lld,send_msg: %s", m_req_id,msg.c_str());

    m_reqtoken = std::to_string(m_req_id);  //暂存发送的token
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_send_timestamp = get_timestamp();
        m_is_receiving = true;
        wait_recv_finished();
        rep_len = m_rsp_len;
    }
    else {
        rep_len = 0;
        CLogger::getInstance()->LOG_E("ccoRouter acq data[%s] error=%d.", topic.c_str(), ret);
    }
    m_thread_notify.unlock();

    return rep_len;
}

int HPLCProxy::clear_hplcarchives(uint8_t* rsp, int& rsplen)
{
    m_thread_notify.lock();
    m_req_id++;
    m_rsp = rsp;
    std::string msg = "";
    std::string topic = HPLC_CLEARARCHIVES_RSP_TOPIC;

    m_hpm.get_hplc_cleararchives_reqmsg(m_req_id, msg);
    CLogger::getInstance()->log_i("clear_hplcarchives  send: %s", msg.c_str());
    int ret = m_mosquitto_manager->publish(topic.c_str(), msg.length(), msg.c_str());
    if (ret == MOSQ_ERR_SUCCESS){
        m_is_receiving = true;
        wait_recv_finished();
    }
    else {
        rsplen = 0;
        CLogger::getInstance()->LOG_E("ccoRouter clear_hplcarchives[%s] error=%d.", topic.c_str(), ret);
    }
    m_thread_notify.unlock();

    return rsplen;
}

void HPLCProxy::OutputDecodedata(char* pData, int len)
{
    char  szInfo[1024] = { 0 };
    char  szch[8] = { 0 };
    for (int i = 0; i < len; i++){
        sprintf(szch, "%02X ", pData[i]);
        strcat(szInfo, szch);
    }

    CLogger::getInstance()->log_d("recv decode data:%s", szInfo);
}